fix(ci): note heartbeats under the run lock and forget the poll record at the ack

Heartbeat now does the acknowledgement and the note of the heartbeat under the
lock of the run, so a heartbeat can no longer be recorded for a job that
ended meanwhile. The record of a poll (which holds the plaintext job token for
the retry of a lost response) is dropped as soon as the assignment is
acknowledged or cancelled. Tests: stale assignment under the lock, the typed
nil, and a retried poll after the acknowledgement.

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

3 files · +171 -7

M internal/app/ci_flow_test.go +20 -0


@@ -627,3 +627,23 @@ jobs:
627	627		require.NoError(t, err)
628	628		assert.Nil(t, again, "a cancelled job is not given again")
629	629		}
	630	+
	631	+	// Once the runner acknowledged a job, a retried poll with the same poll_id has
	632	+	// nothing to repeat: the plaintext token is no longer kept.
	633	+	func TestCIAcknowledgedAssignmentIsNotRepeated(t *testing.T) {
	634	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 150 time.Millisecond })
	635	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	636	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	637	+	var job *pb.Job
	638	+	c.waitFor("a job", func() bool {
	639	+	j, err := runner.fetch("poll-9")
	640	+	require.NoError(t, err)
	641	+	job = j
	642	+	return j != nil
	643	+	})
	644	+	_, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	645	+	require.NoError(t, err)
	646	+	again, err := runner.fetch("poll-9")
	647	+	require.NoError(t, err)
	648	+	assert.Nil(t, again, "the acknowledged assignment is not handed out again")
	649	+	}


@@ -627,3 +627,23 @@ jobs:
627	require.NoError(t, err)	627		require.NoError(t, err)
628	assert.Nil(t, again, "a cancelled job is not given again")	628		assert.Nil(t, again, "a cancelled job is not given again")
629	}	629		}
		630	+
		631	+	// Once the runner acknowledged a job, a retried poll with the same poll_id has
		632	+	// nothing to repeat: the plaintext token is no longer kept.
		633	+	func TestCIAcknowledgedAssignmentIsNotRepeated(t *testing.T) {
		634	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.PollTimeout = 150 time.Millisecond })
		635	+	runner := c.registerRunner("alpha", "ubuntu-latest")
		636	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
		637	+	var job *pb.Job
		638	+	c.waitFor("a job", func() bool {
		639	+	j, err := runner.fetch("poll-9")
		640	+	require.NoError(t, err)
		641	+	job = j
		642	+	return j != nil
		643	+	})
		644	+	_, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
		645	+	require.NoError(t, err)
		646	+	again, err := runner.fetch("poll-9")
		647	+	require.NoError(t, err)
		648	+	assert.Nil(t, again, "the acknowledged assignment is not handed out again")
		649	+	}

A internal/usecase/ci_runner_unit_test.go +150 -0

@@ -0,0 +1,150 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/mock"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/domain"
	15	+	"bnema.dev/ermite/internal/ports"
	16	+	"bnema.dev/ermite/internal/ports/mocks"
	17	+	"bnema.dev/ermite/internal/usecase"
	18	+	)
	19	+
	20	+	// ciUnit is the CI use cases over mocks, with a clock the test moves.
	21	+	type ciUnit struct {
	22	+	store *mocks.CIStore
	23	+	logs *mocks.CILogStore
	24	+	meta *mocks.RepoMetaStore
	25	+	bus *mocks.EventBus
	26	+	clock *fakeClock
	27	+	ci *usecase.CI
	28	+	}
	29	+
	30	+	type fakeClock struct{ t time.Time }
	31	+
	32	+	func (c *fakeClock) Now() time.Time { return c.t }
	33	+
	34	+	func newCIUnit(t testing.T, tweak func(usecase.CIConfig)) *ciUnit {
	35	+	t.Helper()
	36	+	u := &ciUnit{
	37	+	store: mocks.NewCIStore(t), logs: mocks.NewCILogStore(t), meta: mocks.NewRepoMetaStore(t),
	38	+	bus: mocks.NewEventBus(t), clock: &fakeClock{t: now},
	39	+	}
	40	+	cfg := usecase.CIConfig{
	41	+	Store: u.store, Logs: u.logs, Meta: u.meta, Repos: mocks.NewGitRepos(t), Reader: mocks.NewRepoReader(t),
	42	+	Git: mocks.NewMergeGit(t), Diff: mocks.NewCIGit(t), Bus: u.bus, Clock: u.clock, Owner: "owner", PublicURL: "https://forge.test",
	43	+	}
	44	+	if tweak != nil {
	45	+	tweak(&cfg)
	46	+	}
	47	+	u.ci = usecase.NewCI(context.Background(), cfg)
	48	+	t.Cleanup(func() { u.ci.Close(context.Background()) })
	49	+	return u
	50	+	}
	51	+
	52	+	const (
	53	+	jobSecret = "ermitejob_abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN-_0"
	54	+	jobIDOne = "7-0-0"
	55	+	)
	56	+
	57	+	func ciJob(t *testing.T, phase domain.CIJobPhase, secret string) domain.CIJob {
	58	+	t.Helper()
	59	+	return domain.CIJob{
	60	+	ID: jobIDOne, RunID: 7, Attempt: 1, RunAttempt: 1, Repo: name(t, "proj"), Key: "build", Name: "build",
	61	+	Phase: phase, TokenHash: domain.HashTokenSecret(secret), RunnerID: "r1", RunnerName: "alpha", Planned: true,
	62	+	}
	63	+	}
	64	+
	65	+	// expectToken makes the token lookup find j.
	66	+	func (u *ciUnit) expectToken(secret string, j domain.CIJob) {
	67	+	u.store.EXPECT().FindJobByToken(mock.Anything, domain.HashTokenSecret(secret), mock.Anything).Return(j, nil)
	68	+	}
	69	+
	70	+	// S4: a call whose token was replaced between the lookup and the lock (the
	71	+	// assignment was cancelled and given again) is unauthenticated, for every
	72	+	// call that writes the job.
	73	+	func TestStaleAssignmentIsUnauthenticatedUnderTheLock(t *testing.T) {
	74	+	t.Parallel()
	75	+	replaced := func(j domain.CIJob) domain.CIJob {
	76	+	j.TokenHash = domain.HashTokenSecret("ermitejob_replaced")
	77	+	return j
	78	+	}
	79	+	calls := map[string]func(u *ciUnit) error{
	80	+	"UpdateJob": func(u *ciUnit) error {
	81	+	return u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{
	82	+	Number: 1, State: domain.CIStepInProgress, StartedAt: now,
	83	+	}})
	84	+	},
	85	+	"AppendLog": func(u *ciUnit) error {
	86	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
	87	+	return err
	88	+	},
	89	+	}
	90	+	for label, call := range calls {
	91	+	t.Run(label, func(t *testing.T) {
	92	+	t.Parallel()
	93	+	u := newCIUnit(t, nil)
	94	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	95	+	u.expectToken(jobSecret, j)
	96	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(replaced(j), nil)
	97	+	// No write: the stores and the log expect nothing else.
	98	+	assert.ErrorIs(t, call(u), domain.ErrUnauthenticated)
	99	+	})
	100	+	}
	101	+
	102	+	t.Run("CompleteJob", func(t *testing.T) {
	103	+	t.Parallel()
	104	+	u := newCIUnit(t, nil)
	105	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	106	+	u.expectToken(jobSecret, j)
	107	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
	108	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
	109	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{replaced(j)}, nil)
	110	+	_, err := u.ci.CompleteJob(context.Background(), jobSecret, ports.CompleteJobParams{JobID: j.ID, Conclusion: domain.CISuccess})
	111	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
	112	+	})
	113	+
	114	+	t.Run("an ended job answers with its end, not unauthenticated", func(t *testing.T) {
	115	+	t.Parallel()
	116	+	u := newCIUnit(t, nil)
	117	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	118	+	ended := replaced(j)
	119	+	ended.Phase, ended.Conclusion = domain.CIPhaseEnded, domain.CICancelled
	120	+	u.expectToken(jobSecret, j)
	121	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(ended, nil)
	122	+	err := u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{Number: 1, State: domain.CIStepInProgress, StartedAt: now}})
	123	+	got, ok := domain.IsJobEnded(err)
	124	+	require.True(t, ok, "%v", err)
	125	+	assert.Equal(t, domain.CICancelled, got.Conclusion)
	126	+	})
	127	+	}
	128	+
	129	+	// S5: the job vanished while the log limit was handled: a plain not found, never
	130	+	// a typed nil error.
	131	+	func TestLogLimitOnAVanishedJobIsNotFound(t *testing.T) {
	132	+	t.Parallel()
	133	+	u := newCIUnit(t, nil)
	134	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	135	+	u.expectToken(jobSecret, j)
	136	+	// UpdateJob-style lookup under lock, then a log at the limit.
	137	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
	138	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(0, nil)
	139	+	u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(0), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
	140	+	// stopForLogLimit loads the run, and the job is no longer in it.
	141	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
	142	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
	143	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return(nil, nil)
	144	+
	145	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
	146	+	require.Error(t, err)
	147	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	148	+	_, isEnded := domain.IsJobEnded(err)
	149	+	assert.False(t, isEnded)
	150	+	}

@@ -0,0 +1,150 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/mock"
	12	+	"github.com/stretchr/testify/require"
	13	+
	14	+	"bnema.dev/ermite/internal/domain"
	15	+	"bnema.dev/ermite/internal/ports"
	16	+	"bnema.dev/ermite/internal/ports/mocks"
	17	+	"bnema.dev/ermite/internal/usecase"
	18	+	)
	19	+
	20	+	// ciUnit is the CI use cases over mocks, with a clock the test moves.
	21	+	type ciUnit struct {
	22	+	store *mocks.CIStore
	23	+	logs *mocks.CILogStore
	24	+	meta *mocks.RepoMetaStore
	25	+	bus *mocks.EventBus
	26	+	clock *fakeClock
	27	+	ci *usecase.CI
	28	+	}
	29	+
	30	+	type fakeClock struct{ t time.Time }
	31	+
	32	+	func (c *fakeClock) Now() time.Time { return c.t }
	33	+
	34	+	func newCIUnit(t testing.T, tweak func(usecase.CIConfig)) *ciUnit {
	35	+	t.Helper()
	36	+	u := &ciUnit{
	37	+	store: mocks.NewCIStore(t), logs: mocks.NewCILogStore(t), meta: mocks.NewRepoMetaStore(t),
	38	+	bus: mocks.NewEventBus(t), clock: &fakeClock{t: now},
	39	+	}
	40	+	cfg := usecase.CIConfig{
	41	+	Store: u.store, Logs: u.logs, Meta: u.meta, Repos: mocks.NewGitRepos(t), Reader: mocks.NewRepoReader(t),
	42	+	Git: mocks.NewMergeGit(t), Diff: mocks.NewCIGit(t), Bus: u.bus, Clock: u.clock, Owner: "owner", PublicURL: "https://forge.test",
	43	+	}
	44	+	if tweak != nil {
	45	+	tweak(&cfg)
	46	+	}
	47	+	u.ci = usecase.NewCI(context.Background(), cfg)
	48	+	t.Cleanup(func() { u.ci.Close(context.Background()) })
	49	+	return u
	50	+	}
	51	+
	52	+	const (
	53	+	jobSecret = "ermitejob_abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN-_0"
	54	+	jobIDOne = "7-0-0"
	55	+	)
	56	+
	57	+	func ciJob(t *testing.T, phase domain.CIJobPhase, secret string) domain.CIJob {
	58	+	t.Helper()
	59	+	return domain.CIJob{
	60	+	ID: jobIDOne, RunID: 7, Attempt: 1, RunAttempt: 1, Repo: name(t, "proj"), Key: "build", Name: "build",
	61	+	Phase: phase, TokenHash: domain.HashTokenSecret(secret), RunnerID: "r1", RunnerName: "alpha", Planned: true,
	62	+	}
	63	+	}
	64	+
	65	+	// expectToken makes the token lookup find j.
	66	+	func (u *ciUnit) expectToken(secret string, j domain.CIJob) {
	67	+	u.store.EXPECT().FindJobByToken(mock.Anything, domain.HashTokenSecret(secret), mock.Anything).Return(j, nil)
	68	+	}
	69	+
	70	+	// S4: a call whose token was replaced between the lookup and the lock (the
	71	+	// assignment was cancelled and given again) is unauthenticated, for every
	72	+	// call that writes the job.
	73	+	func TestStaleAssignmentIsUnauthenticatedUnderTheLock(t *testing.T) {
	74	+	t.Parallel()
	75	+	replaced := func(j domain.CIJob) domain.CIJob {
	76	+	j.TokenHash = domain.HashTokenSecret("ermitejob_replaced")
	77	+	return j
	78	+	}
	79	+	calls := map[string]func(u *ciUnit) error{
	80	+	"UpdateJob": func(u *ciUnit) error {
	81	+	return u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{
	82	+	Number: 1, State: domain.CIStepInProgress, StartedAt: now,
	83	+	}})
	84	+	},
	85	+	"AppendLog": func(u *ciUnit) error {
	86	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
	87	+	return err
	88	+	},
	89	+	}
	90	+	for label, call := range calls {
	91	+	t.Run(label, func(t *testing.T) {
	92	+	t.Parallel()
	93	+	u := newCIUnit(t, nil)
	94	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	95	+	u.expectToken(jobSecret, j)
	96	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(replaced(j), nil)
	97	+	// No write: the stores and the log expect nothing else.
	98	+	assert.ErrorIs(t, call(u), domain.ErrUnauthenticated)
	99	+	})
	100	+	}
	101	+
	102	+	t.Run("CompleteJob", func(t *testing.T) {
	103	+	t.Parallel()
	104	+	u := newCIUnit(t, nil)
	105	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	106	+	u.expectToken(jobSecret, j)
	107	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
	108	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
	109	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{replaced(j)}, nil)
	110	+	_, err := u.ci.CompleteJob(context.Background(), jobSecret, ports.CompleteJobParams{JobID: j.ID, Conclusion: domain.CISuccess})
	111	+	assert.ErrorIs(t, err, domain.ErrUnauthenticated)
	112	+	})
	113	+
	114	+	t.Run("an ended job answers with its end, not unauthenticated", func(t *testing.T) {
	115	+	t.Parallel()
	116	+	u := newCIUnit(t, nil)
	117	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	118	+	ended := replaced(j)
	119	+	ended.Phase, ended.Conclusion = domain.CIPhaseEnded, domain.CICancelled
	120	+	u.expectToken(jobSecret, j)
	121	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(ended, nil)
	122	+	err := u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{Number: 1, State: domain.CIStepInProgress, StartedAt: now}})
	123	+	got, ok := domain.IsJobEnded(err)
	124	+	require.True(t, ok, "%v", err)
	125	+	assert.Equal(t, domain.CICancelled, got.Conclusion)
	126	+	})
	127	+	}
	128	+
	129	+	// S5: the job vanished while the log limit was handled: a plain not found, never
	130	+	// a typed nil error.
	131	+	func TestLogLimitOnAVanishedJobIsNotFound(t *testing.T) {
	132	+	t.Parallel()
	133	+	u := newCIUnit(t, nil)
	134	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	135	+	u.expectToken(jobSecret, j)
	136	+	// UpdateJob-style lookup under lock, then a log at the limit.
	137	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
	138	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(0, nil)
	139	+	u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(0), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
	140	+	// stopForLogLimit loads the run, and the job is no longer in it.
	141	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
	142	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
	143	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return(nil, nil)
	144	+
	145	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
	146	+	require.Error(t, err)
	147	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	148	+	_, isEnded := domain.IsJobEnded(err)
	149	+	assert.False(t, isEnded)
	150	+	}

M internal/usecase/ci_watch.go +1 -7


@@ -191,13 +191,7 @@ func (c CI) cancelAssignment(ctx context.Context, rs runState, j *domain.CIJob
191	191		j.TokenHash, j.TokenValidUntil = domain.TokenHash{}, time.Time{}
192	192		j.RunnerID, j.RunnerName, j.Isolation = "", "", domain.CIIsolationNone
193	193		j.Steps, j.Outputs, j.AssignedAt = nil, nil, time.Time{}
194		-	c.mu.Lock()
195		-	for id, p := range c.polls {
196		-	if p.ref == jobRef(*j) {
197		-	delete(c.polls, id)
198		-	}
199		-	}
200		-	c.mu.Unlock()
	194	+	c.forgetPoll(jobRef(*j))
201	195		if j.CancelReason != domain.CICancelNone {
202	196		// Cancelled (by the owner, fail-fast or the time budget) before the
203	197		// runner answered: it is over, not queued again for another runner.

 	j.TokenHash, j.TokenValidUntil = domain.TokenHash{}, time.Time{}
 	j.RunnerID, j.RunnerName, j.Isolation = "", "", domain.CIIsolationNone
 	j.Steps, j.Outputs, j.AssignedAt = nil, nil, time.Time{}
-	c.mu.Lock()
+	c.forgetPoll(jobRef(*j))
-	for id, p := range c.polls {
-		if p.ref == jobRef(*j) {
-			delete(c.polls, id)
-		}
-	}
-	c.mu.Unlock()
 	if j.CancelReason != domain.CICancelNone {
 		// Cancelled (by the owner, fail-fast or the time budget) before the
 		// runner answered: it is over, not queued again for another runner.