fix(ci): check the assignment under the run lock and never return a typed nil

UpdateJob, AppendLog and CompleteJob compared the token only before taking the
lock, so a call from a runner whose assignment was cancelled and given again
could still write the job. They now compare it under the lock, like
Heartbeat, unless the job has ended. stopForLogLimit returned a nil
*JobEndedError as an error when the job had vanished: it is now not found.

commit: 8d95a4923a16f83622b5feec991c3b481d3f70e4
author: brice · 9d ago · 2026-10-12 19:11 UTC
parent: 9728ecc
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +65 -27

M internal/usecase/ci_runner.go +65 -27


@@ -303,9 +303,23 @@ func (c CI) lockedJob(ctx context.Context, j domain.CIJob, fn func(cur domain.
303	303		if err != nil {
304	304		return err
305	305		}
	306	+	if err := sameAssignment(j, cur); err != nil {
	307	+	return err
	308	+	}
306	309		return fn(&cur)
307	310		}
308	311
	312	+	// sameAssignment checks, under the lock of the run, that the job a token was
	313	+	// found for is still given under that token: an assignment cancelled and given
	314	+	// again since has a new one, and the old runner is unauthenticated. A job that
	315	+	// has ended answers with its end instead (the callers do).
	316	+	func sameAssignment(was, cur domain.CIJob) error {
	317	+	if !cur.Ended() && cur.TokenHash != was.TokenHash {
	318	+	return domain.ErrUnauthenticated
	319	+	}
	320	+	return nil
	321	+	}
	322	+
309	323		// saveJob writes one job (its steps changed) and publishes JobUpdated. The
310	324		// status of the run does not depend on steps.
311	325		func (c CI) saveJob(ctx context.Context, j domain.CIJob) error {

@@ -562,12 +576,16 @@ func (c *CI) stopForLogLimit(ctx context.Context, j domain.CIJob) error {
562	576		var ended *domain.JobEndedError
563	577		err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
564	578		cur := rs.find(j.ID)
565		-	if cur == nil \|\| cur.Ended() {
566		-	if cur != nil {
567		-	ended = &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}
568		-	}
	579	+	if cur == nil {
	580	+	return domain.ErrNotFound
	581	+	}
	582	+	if cur.Ended() {
	583	+	ended = &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}
569	584		return nil
570	585		}
	586	+	if err := sameAssignment(j, *cur); err != nil {
	587	+	return err
	588	+	}
571	589		_, _ = c.cfg.Logs.AppendForced(ctx, logRef(*cur), []domain.CILogLine{{Time: c.now(), Text: "##[error]" + domain.CILogLimitMessage}})
572	590		rs.end(cur, domain.CIFailure, domain.CILogLimitMessage)
573	591		ended = &domain.JobEndedError{Conclusion: domain.CIFailure, Message: domain.CILogLimitMessage}

@@ -579,6 +597,9 @@ func (c *CI) stopForLogLimit(ctx context.Context, j domain.CIJob) error {
579	597		if err != nil {
580	598		return err
581	599		}
	600	+	if ended == nil {
	601	+	return domain.ErrNotFound // a typed nil would be taken for an error
	602	+	}
582	603		return ended
583	604		}
584	605

@@ -592,33 +613,47 @@ func (c *CI) Heartbeat(ctx context.Context, secret, jobID string) error {
592	613		if j.Ended() {
593	614		return endedError(j)
594	615		}
595		-	now := c.now()
596		-	if j.Phase == domain.CIPhaseAssigned {
597		-	err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
598		-	cur := rs.find(j.ID)
599		-	switch {
600		-	case cur == nil:
601		-	return domain.ErrUnauthenticated
602		-	case cur.Ended():
603		-	return endedError(*cur)
604		-	case cur.TokenHash != j.TokenHash:
605		-	return domain.ErrUnauthenticated // the assignment was cancelled and given again
606		-	case cur.Phase != domain.CIPhaseAssigned:
607		-	return nil // acknowledged by a concurrent call
608		-	}
	616	+	// Everything is done under the lock of the run: the acknowledgement, and
	617	+	// the note of the heartbeat, which must not be written once the job ended
	618	+	// (end() forgets it under the same lock).
	619	+	return c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
	620	+	cur := rs.find(j.ID)
	621	+	switch {
	622	+	case cur == nil:
	623	+	return domain.ErrUnauthenticated
	624	+	case cur.Ended():
	625	+	return endedError(*cur)
	626	+	case cur.TokenHash != j.TokenHash:
	627	+	return domain.ErrUnauthenticated // the assignment was cancelled and given again
	628	+	}
	629	+	now := c.now()
	630	+	if cur.Phase == domain.CIPhaseAssigned {
609	631		cur.Phase, cur.AckedAt, cur.StartedAt, cur.AssignCancels = domain.CIPhaseRunning, now, now, 0
610	632		rs.touch(cur)
611		-	return rs.flush(ctx)
612		-	})
613		-	if err != nil {
614		-	return err
	633	+	if err := rs.flush(ctx); err != nil {
	634	+	return err
	635	+	}
	636	+	c.forgetPoll(jobRef(*cur)) // the token it holds is the runner's now
615	637		}
616		-	}
	638	+	c.mu.Lock()
	639	+	c.heartbeats[jobRef(*cur)] = now
	640	+	c.seen[cur.RunnerID] = now
	641	+	c.mu.Unlock()
	642	+	return nil
	643	+	})
	644	+	}
	645	+
	646	+	// forgetPoll drops the remembered assignment (and its plaintext token) of a
	647	+	// job: after the acknowledgement or the cancel of the assignment, a retried
	648	+	// poll has nothing to repeat.
	649	+	func (c *CI) forgetPoll(ref string) {
617	650		c.mu.Lock()
618		-	c.heartbeats[jobRef(j)] = now
619		-	c.seen[j.RunnerID] = now
620		-	c.mu.Unlock()
621		-	return nil
	651	+	defer c.mu.Unlock()
	652	+	for id, p := range c.polls {
	653	+	if p.ref == ref {
	654	+	delete(c.polls, id)
	655	+	}
	656	+	}
622	657		}
623	658
624	659		// WatchJob streams what the runner must react to: the cancel request (the

@@ -705,6 +740,9 @@ func (c *CI) CompleteJob(ctx context.Context, secret string, params ports.Comple
705	740		res = ports.CompleteJobResult{Conclusion: cur.Conclusion, Error: cur.Error}
706	741		return nil
707	742		}
	743	+	if err := sameAssignment(j, *cur); err != nil {
	744	+	return err
	745	+	}
708	746		steps, _, err := applySteps(cur.Steps, params.Steps)
709	747		if err != nil {
710	748		return err

 	if err != nil {
 		return err
 	}
+	if err := sameAssignment(j, cur); err != nil {
+		return err
+	}
 	return fn(&cur)
 }
+// sameAssignment checks, under the lock of the run, that the job a token was
+// found for is still given under that token: an assignment cancelled and given
+// again since has a new one, and the old runner is unauthenticated. A job that
+// has ended answers with its end instead (the callers do).
+func sameAssignment(was, cur domain.CIJob) error {
+	if !cur.Ended() && cur.TokenHash != was.TokenHash {
+		return domain.ErrUnauthenticated
+	}
+	return nil
+}
 // saveJob writes one job (its steps changed) and publishes JobUpdated. The
 // status of the run does not depend on steps.
 func (c *CI) saveJob(ctx context.Context, j *domain.CIJob) error {
 	var ended *domain.JobEndedError
 	err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
 		cur := rs.find(j.ID)
-		if cur == nil || cur.Ended() {
+		if cur == nil {
-			if cur != nil {
+			return domain.ErrNotFound
-				ended = &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}
+		}
-			}
+		if cur.Ended() {
+			ended = &domain.JobEndedError{Conclusion: cur.Conclusion, Message: cur.Error}
 			return nil
 		}
+		if err := sameAssignment(j, *cur); err != nil {
+			return err
+		}
 		_, _ = c.cfg.Logs.AppendForced(ctx, logRef(*cur), []domain.CILogLine{{Time: c.now(), Text: "##[error]" + domain.CILogLimitMessage}})
 		rs.end(cur, domain.CIFailure, domain.CILogLimitMessage)
 		ended = &domain.JobEndedError{Conclusion: domain.CIFailure, Message: domain.CILogLimitMessage}
 	if err != nil {
 		return err
 	}
+	if ended == nil {
+		return domain.ErrNotFound // a typed nil would be taken for an error
+	}
 	return ended
 }
 	if j.Ended() {
 		return endedError(j)
 	}
-	now := c.now()
+	// Everything is done under the lock of the run: the acknowledgement, and
-	if j.Phase == domain.CIPhaseAssigned {
+	// the note of the heartbeat, which must not be written once the job ended
-		err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
+	// (end() forgets it under the same lock).
-			cur := rs.find(j.ID)
+	return c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
-			switch {
+		cur := rs.find(j.ID)
-			case cur == nil:
+		switch {
-				return domain.ErrUnauthenticated
+		case cur == nil:
-			case cur.Ended():
+			return domain.ErrUnauthenticated
-				return endedError(*cur)
+		case cur.Ended():
-			case cur.TokenHash != j.TokenHash:
+			return endedError(*cur)
-				return domain.ErrUnauthenticated // the assignment was cancelled and given again
+		case cur.TokenHash != j.TokenHash:
-			case cur.Phase != domain.CIPhaseAssigned:
+			return domain.ErrUnauthenticated // the assignment was cancelled and given again
-				return nil // acknowledged by a concurrent call
+		}
-			}
+		now := c.now()
+		if cur.Phase == domain.CIPhaseAssigned {
 			cur.Phase, cur.AckedAt, cur.StartedAt, cur.AssignCancels = domain.CIPhaseRunning, now, now, 0
 			rs.touch(cur)
-			return rs.flush(ctx)
+			if err := rs.flush(ctx); err != nil {
-		})
+				return err
-		if err != nil {
+			}
-			return err
+			c.forgetPoll(jobRef(*cur)) // the token it holds is the runner's now
 		}
-	}
+		c.mu.Lock()
+		c.heartbeats[jobRef(*cur)] = now
+		c.seen[cur.RunnerID] = now
+		c.mu.Unlock()
+		return nil
+	})
+}
+// forgetPoll drops the remembered assignment (and its plaintext token) of a
+// job: after the acknowledgement or the cancel of the assignment, a retried
+// poll has nothing to repeat.
+func (c *CI) forgetPoll(ref string) {
 	c.mu.Lock()
-	c.heartbeats[jobRef(j)] = now
+	defer c.mu.Unlock()
-	c.seen[j.RunnerID] = now
+	for id, p := range c.polls {
-	c.mu.Unlock()
+		if p.ref == ref {
-	return nil
+			delete(c.polls, id)
+		}
+	}
 }
 // WatchJob streams what the runner must react to: the cancel request (the
 			res = ports.CompleteJobResult{Conclusion: cur.Conclusion, Error: cur.Error}
 			return nil
 		}
+		if err := sameAssignment(j, *cur); err != nil {
+			return err
+		}
 		steps, _, err := applySteps(cur.Steps, params.Steps)
 		if err != nil {
 			return err