fix(ci): log every failure of the sweep, and test the sweep, the log limit and GetJob

The sweep dropped the errors of the log directory listing and of the orphan
removal; they are logged now, and a run whose lookup fails keeps its logs. Tests
with mocks and a clock that moves: old runs and orphan logs removed, a store that
fails, a RemoveRuns that fails, a failing listing, a lookup error; the log limit
ends the job with the forced last line; the rpc GetJob conversion and the
JobEnded detail.

commit: 916e8496823dbaf81fd2c6af255c1812dc8a963a
author: brice · 9d ago · 2026-10-12 19:22 UTC
parent: cd99c65
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +249 -8

A internal/adapters/rpc/ci_test.go +101 -0

@@ -0,0 +1,101 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"connectrpc.com/connect"
	11	+	"github.com/stretchr/testify/assert"
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	pb "bnema.dev/ermite-api/gen/ermite"
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	)
	20	+
	21	+	// RunService.GetJob: the job with its steps and the range of their lines, the
	22	+	// states and conclusions of the contract, and the codes of the refusals.
	23	+	func TestRunServiceGetJob(t *testing.T) {
	24	+	t.Parallel()
	25	+	at := time.Date(2026, 10, 12, 10, 0, 0, 0, time.UTC)
	26	+	job := ports.JobView{CIJob: domain.CIJob{
	27	+	ID: "7-0-0", RunID: 7, Attempt: 2, RunAttempt: 2, Repo: mustRepoName(t, "demo"), Key: "build", Name: "build (linux)",
	28	+	Phase: domain.CIPhaseEnded, Conclusion: domain.CIFailure, Error: "boom", Labels: []string{"self-hosted"}, Needs: []string{"lint"},
	29	+	RunnerName: "alpha", Isolation: domain.CIIsolationVM, CreatedAt: at, StartedAt: at, CompletedAt: at.Add(time.Minute), Version: 9,
	30	+	Steps: []domain.CIStep{
	31	+	{Number: 1, Name: "Set up job", State: domain.CIStepCompleted, Conclusion: domain.CISuccess, StartedAt: at, CompletedAt: at, HasLines: true, LogStart: 0},
	32	+	{Number: 2, Name: "Run make", State: domain.CIStepCompleted, Conclusion: domain.CIFailure, StartedAt: at, CompletedAt: at, HasLines: true, LogStart: 3},
	33	+	},
	34	+	}, Lines: 8}
	35	+
	36	+	runs := mocks.NewRunService(t)
	37	+	runs.EXPECT().GetJob(mock.Anything, domain.Owner(), mustRepoName(t, "demo"), "7-0-0", int32(2)).Return(job, nil).Once()
	38	+	resp, err := (&runService{runs: runs}).GetJob(ownerCtx(), connect.NewRequest(&pb.GetJobRequest{Repo: "demo", JobId: "7-0-0", Attempt: 2}))
	39	+	require.NoError(t, err)
	40	+	got := resp.Msg.GetJob()
	41	+	assert.Equal(t, "7-0-0", got.GetId())
	42	+	assert.EqualValues(t, 2, got.GetAttempt())
	43	+	assert.Equal(t, "build (linux)", got.GetName())
	44	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, got.GetStatus())
	45	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, got.GetConclusion())
	46	+	assert.Equal(t, "boom", got.GetError())
	47	+	assert.Equal(t, pb.RunnerIsolation_RUNNER_ISOLATION_VM, got.GetIsolation())
	48	+	assert.Equal(t, []string{"lint"}, got.GetNeeds())
	49	+	assert.EqualValues(t, 9, got.GetVersion())
	50	+	require.Len(t, got.GetSteps(), 2)
	51	+	assert.Equal(t, []int64{0, 3}, []int64{got.GetSteps()[0].GetLogStart(), got.GetSteps()[0].GetLogEnd()}, "a step ends where the next begins")
	52	+	assert.Equal(t, []int64{3, 8}, []int64{got.GetSteps()[1].GetLogStart(), got.GetSteps()[1].GetLogEnd()}, "the last ends at the end of the log")
	53	+	assert.Equal(t, pb.StepState_STEP_STATE_COMPLETED, got.GetSteps()[1].GetState())
	54	+
	55	+	for name, tc := range map[string]struct {
	56	+	repo string
	57	+	err error
	58	+	code connect.Code
	59	+	}{
	60	+	"invalid repo": {"../x", nil, connect.CodeInvalidArgument},
	61	+	"unknown job": {"demo", domain.ErrNotFound, connect.CodeNotFound},
	62	+	"not readable": {"demo", domain.ErrNotFound, connect.CodeNotFound},
	63	+	"store failure": {"demo", assert.AnError, connect.CodeInternal},
	64	+	} {
	65	+	t.Run(name, func(t *testing.T) {
	66	+	t.Parallel()
	67	+	runs := mocks.NewRunService(t)
	68	+	if tc.err != nil {
	69	+	runs.EXPECT().GetJob(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ports.JobView{}, tc.err).Once()
	70	+	}
	71	+	_, err := (&runService{runs: runs}).GetJob(ownerCtx(), connect.NewRequest(&pb.GetJobRequest{Repo: tc.repo, JobId: "7-0-0"}))
	72	+	requireCode(t, tc.code, err)
	73	+	})
	74	+	}
	75	+	}
	76	+
	77	+	// The end of a job reaches the runner as failed_precondition with the JobEnded
	78	+	// detail; a bad credential counts against the client address.
	79	+	func TestRunnerServiceErrors(t *testing.T) {
	80	+	t.Parallel()
	81	+	ended := &domain.JobEndedError{Conclusion: domain.CICancelled, Message: "stopped"}
	82	+	err := ciError(ownerCtx(), ended)
	83	+	requireCode(t, connect.CodeFailedPrecondition, err)
	84	+	var ce *connect.Error
	85	+	require.ErrorAs(t, err, &ce)
	86	+	require.Len(t, ce.Details(), 1)
	87	+	msg, derr := ce.Details()[0].Value()
	88	+	require.NoError(t, derr)
	89	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, msg.(*pb.JobEnded).GetConclusion())
	90	+	assert.Equal(t, "stopped", msg.(*pb.JobEnded).GetError())
	91	+
	92	+	requireCode(t, connect.CodeOutOfRange, ciError(ownerCtx(), &ports.ErrCILogGap{Count: 4}))
	93	+
	94	+	limiter := mocks.NewFailureLimiter(t)
	95	+	limiter.EXPECT().Fail(mock.Anything, "client-key").Return().Once()
	96	+	s := &runnerService{runners: mocks.NewRunnerService(t), limiter: limiter}
	97	+	ctx := t.Context()
	98	+	ctx = context.WithValue(ctx, addrKey{}, "client-key")
	99	+	requireCode(t, connect.CodeUnauthenticated, s.failed(ctx, domain.ErrUnauthenticated))
	100	+	requireCode(t, connect.CodePermissionDenied, s.failed(ctx, domain.ErrForbidden)) // not a bad credential: no Fail
	101	+	}

@@ -0,0 +1,101 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package rpc
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"connectrpc.com/connect"
	11	+	"github.com/stretchr/testify/assert"
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+
	15	+	pb "bnema.dev/ermite-api/gen/ermite"
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	)
	20	+
	21	+	// RunService.GetJob: the job with its steps and the range of their lines, the
	22	+	// states and conclusions of the contract, and the codes of the refusals.
	23	+	func TestRunServiceGetJob(t *testing.T) {
	24	+	t.Parallel()
	25	+	at := time.Date(2026, 10, 12, 10, 0, 0, 0, time.UTC)
	26	+	job := ports.JobView{CIJob: domain.CIJob{
	27	+	ID: "7-0-0", RunID: 7, Attempt: 2, RunAttempt: 2, Repo: mustRepoName(t, "demo"), Key: "build", Name: "build (linux)",
	28	+	Phase: domain.CIPhaseEnded, Conclusion: domain.CIFailure, Error: "boom", Labels: []string{"self-hosted"}, Needs: []string{"lint"},
	29	+	RunnerName: "alpha", Isolation: domain.CIIsolationVM, CreatedAt: at, StartedAt: at, CompletedAt: at.Add(time.Minute), Version: 9,
	30	+	Steps: []domain.CIStep{
	31	+	{Number: 1, Name: "Set up job", State: domain.CIStepCompleted, Conclusion: domain.CISuccess, StartedAt: at, CompletedAt: at, HasLines: true, LogStart: 0},
	32	+	{Number: 2, Name: "Run make", State: domain.CIStepCompleted, Conclusion: domain.CIFailure, StartedAt: at, CompletedAt: at, HasLines: true, LogStart: 3},
	33	+	},
	34	+	}, Lines: 8}
	35	+
	36	+	runs := mocks.NewRunService(t)
	37	+	runs.EXPECT().GetJob(mock.Anything, domain.Owner(), mustRepoName(t, "demo"), "7-0-0", int32(2)).Return(job, nil).Once()
	38	+	resp, err := (&runService{runs: runs}).GetJob(ownerCtx(), connect.NewRequest(&pb.GetJobRequest{Repo: "demo", JobId: "7-0-0", Attempt: 2}))
	39	+	require.NoError(t, err)
	40	+	got := resp.Msg.GetJob()
	41	+	assert.Equal(t, "7-0-0", got.GetId())
	42	+	assert.EqualValues(t, 2, got.GetAttempt())
	43	+	assert.Equal(t, "build (linux)", got.GetName())
	44	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, got.GetStatus())
	45	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, got.GetConclusion())
	46	+	assert.Equal(t, "boom", got.GetError())
	47	+	assert.Equal(t, pb.RunnerIsolation_RUNNER_ISOLATION_VM, got.GetIsolation())
	48	+	assert.Equal(t, []string{"lint"}, got.GetNeeds())
	49	+	assert.EqualValues(t, 9, got.GetVersion())
	50	+	require.Len(t, got.GetSteps(), 2)
	51	+	assert.Equal(t, []int64{0, 3}, []int64{got.GetSteps()[0].GetLogStart(), got.GetSteps()[0].GetLogEnd()}, "a step ends where the next begins")
	52	+	assert.Equal(t, []int64{3, 8}, []int64{got.GetSteps()[1].GetLogStart(), got.GetSteps()[1].GetLogEnd()}, "the last ends at the end of the log")
	53	+	assert.Equal(t, pb.StepState_STEP_STATE_COMPLETED, got.GetSteps()[1].GetState())
	54	+
	55	+	for name, tc := range map[string]struct {
	56	+	repo string
	57	+	err error
	58	+	code connect.Code
	59	+	}{
	60	+	"invalid repo": {"../x", nil, connect.CodeInvalidArgument},
	61	+	"unknown job": {"demo", domain.ErrNotFound, connect.CodeNotFound},
	62	+	"not readable": {"demo", domain.ErrNotFound, connect.CodeNotFound},
	63	+	"store failure": {"demo", assert.AnError, connect.CodeInternal},
	64	+	} {
	65	+	t.Run(name, func(t *testing.T) {
	66	+	t.Parallel()
	67	+	runs := mocks.NewRunService(t)
	68	+	if tc.err != nil {
	69	+	runs.EXPECT().GetJob(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ports.JobView{}, tc.err).Once()
	70	+	}
	71	+	_, err := (&runService{runs: runs}).GetJob(ownerCtx(), connect.NewRequest(&pb.GetJobRequest{Repo: tc.repo, JobId: "7-0-0"}))
	72	+	requireCode(t, tc.code, err)
	73	+	})
	74	+	}
	75	+	}
	76	+
	77	+	// The end of a job reaches the runner as failed_precondition with the JobEnded
	78	+	// detail; a bad credential counts against the client address.
	79	+	func TestRunnerServiceErrors(t *testing.T) {
	80	+	t.Parallel()
	81	+	ended := &domain.JobEndedError{Conclusion: domain.CICancelled, Message: "stopped"}
	82	+	err := ciError(ownerCtx(), ended)
	83	+	requireCode(t, connect.CodeFailedPrecondition, err)
	84	+	var ce *connect.Error
	85	+	require.ErrorAs(t, err, &ce)
	86	+	require.Len(t, ce.Details(), 1)
	87	+	msg, derr := ce.Details()[0].Value()
	88	+	require.NoError(t, derr)
	89	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, msg.(*pb.JobEnded).GetConclusion())
	90	+	assert.Equal(t, "stopped", msg.(*pb.JobEnded).GetError())
	91	+
	92	+	requireCode(t, connect.CodeOutOfRange, ciError(ownerCtx(), &ports.ErrCILogGap{Count: 4}))
	93	+
	94	+	limiter := mocks.NewFailureLimiter(t)
	95	+	limiter.EXPECT().Fail(mock.Anything, "client-key").Return().Once()
	96	+	s := &runnerService{runners: mocks.NewRunnerService(t), limiter: limiter}
	97	+	ctx := t.Context()
	98	+	ctx = context.WithValue(ctx, addrKey{}, "client-key")
	99	+	requireCode(t, connect.CodeUnauthenticated, s.failed(ctx, domain.ErrUnauthenticated))
	100	+	requireCode(t, connect.CodePermissionDenied, s.failed(ctx, domain.ErrForbidden)) // not a bad credential: no Fail
	101	+	}

M internal/usecase/ci_runner_unit_test.go +127 -0


@@ -3,10 +3,13 @@
3	3		package usecase_test
4	4
5	5		import (
	6	+	"bytes"
6	7		"context"
	8	+	"errors"
7	9		"testing"
8	10		"time"
9	11
	12	+	"github.com/bnema/zerowrap"
10	13		"github.com/stretchr/testify/assert"
11	14		"github.com/stretchr/testify/mock"
12	15		"github.com/stretchr/testify/require"

@@ -285,3 +288,127 @@ func TestCommitChecksRefusesATruncatedList(t *testing.T) {
285	288		require.NoError(t, err)
286	289		assert.Len(t, checks, 1)
287	290		}
	291	+
	292	+	// logCapture is a context whose logger writes into a buffer the test reads.
	293	+	func logCapture(t testing.T) (context.Context, bytes.Buffer) {
	294	+	t.Helper()
	295	+	var buf bytes.Buffer
	296	+	log := zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &buf})
	297	+	return zerowrap.WithCtx(context.Background(), log), &buf
	298	+	}
	299	+
	300	+	// M5, M6: the sweep removes the runs older than the retention (with the clock
	301	+	// moved, not waited for), the logs of runs no record has any more, and says so
	302	+	// when it cannot.
	303	+	func TestSweep(t *testing.T) {
	304	+	t.Parallel()
	305	+	retention := 30 * 24 * time.Hour
	306	+
	307	+	t.Run("old runs go with their logs, orphan logs too", func(t *testing.T) {
	308	+	t.Parallel()
	309	+	u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Retention = retention })
	310	+	u.clock.t = now.Add(100 * 24 * time.Hour)
	311	+	ctx, out := logCapture(t)
	312	+	u.store.EXPECT().Sweep(mock.Anything, u.clock.t.Add(-retention)).Return([]int64{3, 4}, nil)
	313	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3, 4}).Return(nil).Once()
	314	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{5, 6, 7}, nil)
	315	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(5)).Return(domain.CIRunSource{}, nil)
	316	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(6)).Return(domain.CIRunSource{}, domain.ErrNotFound)
	317	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(7)).Return(domain.CIRunSource{}, domain.ErrNotFound)
	318	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{6, 7}).Return(nil).Once()
	319	+	u.ci.Sweep(ctx)
	320	+	assert.NotContains(t, out.String(), `"level":"error"`)
	321	+	assert.Contains(t, out.String(), "old runs removed")
	322	+	})
	323	+
	324	+	t.Run("a store that fails ends the sweep, logged", func(t *testing.T) {
	325	+	t.Parallel()
	326	+	u := newCIUnit(t, nil)
	327	+	ctx, out := logCapture(t)
	328	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, errors.New("kv closed"))
	329	+	u.ci.Sweep(ctx) // the log store expects no call
	330	+	assert.Contains(t, out.String(), "sweep the old runs")
	331	+	assert.Contains(t, out.String(), "kv closed")
	332	+	})
	333	+
	334	+	t.Run("a log directory that cannot be removed is logged and the orphans are still looked for", func(t *testing.T) {
	335	+	t.Parallel()
	336	+	u := newCIUnit(t, nil)
	337	+	ctx, out := logCapture(t)
	338	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return([]int64{3}, nil)
	339	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3}).Return(errors.New("permission denied")).Once()
	340	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
	341	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, domain.ErrNotFound)
	342	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{9}).Return(errors.New("disk on fire")).Once()
	343	+	u.ci.Sweep(ctx)
	344	+	assert.Contains(t, out.String(), "remove the logs of the old runs")
	345	+	assert.Contains(t, out.String(), "permission denied")
	346	+	assert.Contains(t, out.String(), "remove the logs of runs that no longer exist")
	347	+	assert.Contains(t, out.String(), "disk on fire")
	348	+	})
	349	+
	350	+	t.Run("a listing of the log directories that fails is logged", func(t *testing.T) {
	351	+	t.Parallel()
	352	+	u := newCIUnit(t, nil)
	353	+	ctx, out := logCapture(t)
	354	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
	355	+	u.logs.EXPECT().RunIDs(mock.Anything).Return(nil, errors.New("io error"))
	356	+	u.ci.Sweep(ctx)
	357	+	assert.Contains(t, out.String(), "list the log directories")
	358	+	})
	359	+
	360	+	t.Run("a run with a lookup error keeps its logs", func(t *testing.T) {
	361	+	t.Parallel()
	362	+	u := newCIUnit(t, nil)
	363	+	ctx, out := logCapture(t)
	364	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
	365	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
	366	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, errors.New("kv closed"))
	367	+	u.ci.Sweep(ctx) // RemoveRuns is not expected: an unknown state removes nothing
	368	+	assert.Contains(t, out.String(), "look up a run")
	369	+	})
	370	+	}
	371	+
	372	+	// M5: a log that would pass its limit appends nothing; the server adds a last
	373	+	// line saying so, fails the job and answers failed_precondition with the end.
	374	+	func TestLogLimitEndsTheJob(t *testing.T) {
	375	+	t.Parallel()
	376	+	u := newCIUnit(t, nil)
	377	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
	378	+	j.Steps = []domain.CIStep{{Number: 1, State: domain.CIStepInProgress, StartedAt: now, HasLines: true}}
	379	+	u.expectToken(jobSecret, j)
	380	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
	381	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(10, nil)
	382	+	u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(10), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
	383	+
	384	+	// stopForLogLimit: the run is loaded, the last line is forced in, the job ends.
	385	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo, WorkflowYAML: "on: push\njobs:\n build:\n runs-on: x\n steps:\n - run: echo\n"}, nil)
	386	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1, WorkflowPath: "w.yml"}, nil)
	387	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{j}, nil)
	388	+	var forced []domain.CILogLine
	389	+	u.logs.EXPECT().AppendForced(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
	390	+	func(_ context.Context, _ ports.CILogRef, lines []domain.CILogLine) (int64, error) {
	391	+	forced = lines
	392	+	return 11, nil
	393	+	})
	394	+	var written ports.CIWrite
	395	+	u.store.EXPECT().Apply(mock.Anything, mock.Anything).RunAndReturn(
	396	+	func(_ context.Context, w ports.CIWrite) (uint64, error) { written = w; return 5, nil })
	397	+	u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPrivate}, nil)
	398	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(11, nil).Maybe()
	399	+	u.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
	400	+
	401	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 10, []domain.CILogLine{{Time: now, Text: "one more"}})
	402	+	ended, ok := domain.IsJobEnded(err)
	403	+	require.True(t, ok, "%v", err)
	404	+	assert.Equal(t, domain.CIFailure, ended.Conclusion)
	405	+	assert.Contains(t, ended.Message, "512 MiB")
	406	+
	407	+	require.Len(t, forced, 1)
	408	+	assert.Contains(t, forced[0].Text, "##[error]")
	409	+	assert.Contains(t, forced[0].Text, "512 MiB")
	410	+	require.Len(t, written.Jobs, 1)
	411	+	assert.Equal(t, domain.CIPhaseEnded, written.Jobs[0].Phase)
	412	+	assert.Equal(t, domain.CIFailure, written.Jobs[0].Conclusion)
	413	+	assert.Equal(t, domain.CIStepCompleted, written.Jobs[0].Steps[0].State, "an open step is completed with the job")
	414	+	}


@@ -3,10 +3,13 @@
3	package usecase_test	3		package usecase_test
4		4
5	import (	5		import (
		6	+	"bytes"
6	"context"	7		"context"
		8	+	"errors"
7	"testing"	9		"testing"
8	"time"	10		"time"
9		11
		12	+	"github.com/bnema/zerowrap"
10	"github.com/stretchr/testify/assert"	13		"github.com/stretchr/testify/assert"
11	"github.com/stretchr/testify/mock"	14		"github.com/stretchr/testify/mock"
12	"github.com/stretchr/testify/require"	15		"github.com/stretchr/testify/require"

@@ -285,3 +288,127 @@ func TestCommitChecksRefusesATruncatedList(t *testing.T) {
285	require.NoError(t, err)	288		require.NoError(t, err)
286	assert.Len(t, checks, 1)	289		assert.Len(t, checks, 1)
287	}	290		}
		291	+
		292	+	// logCapture is a context whose logger writes into a buffer the test reads.
		293	+	func logCapture(t testing.T) (context.Context, bytes.Buffer) {
		294	+	t.Helper()
		295	+	var buf bytes.Buffer
		296	+	log := zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &buf})
		297	+	return zerowrap.WithCtx(context.Background(), log), &buf
		298	+	}
		299	+
		300	+	// M5, M6: the sweep removes the runs older than the retention (with the clock
		301	+	// moved, not waited for), the logs of runs no record has any more, and says so
		302	+	// when it cannot.
		303	+	func TestSweep(t *testing.T) {
		304	+	t.Parallel()
		305	+	retention := 30 * 24 * time.Hour
		306	+
		307	+	t.Run("old runs go with their logs, orphan logs too", func(t *testing.T) {
		308	+	t.Parallel()
		309	+	u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Retention = retention })
		310	+	u.clock.t = now.Add(100 * 24 * time.Hour)
		311	+	ctx, out := logCapture(t)
		312	+	u.store.EXPECT().Sweep(mock.Anything, u.clock.t.Add(-retention)).Return([]int64{3, 4}, nil)
		313	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3, 4}).Return(nil).Once()
		314	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{5, 6, 7}, nil)
		315	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(5)).Return(domain.CIRunSource{}, nil)
		316	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(6)).Return(domain.CIRunSource{}, domain.ErrNotFound)
		317	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(7)).Return(domain.CIRunSource{}, domain.ErrNotFound)
		318	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{6, 7}).Return(nil).Once()
		319	+	u.ci.Sweep(ctx)
		320	+	assert.NotContains(t, out.String(), `"level":"error"`)
		321	+	assert.Contains(t, out.String(), "old runs removed")
		322	+	})
		323	+
		324	+	t.Run("a store that fails ends the sweep, logged", func(t *testing.T) {
		325	+	t.Parallel()
		326	+	u := newCIUnit(t, nil)
		327	+	ctx, out := logCapture(t)
		328	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, errors.New("kv closed"))
		329	+	u.ci.Sweep(ctx) // the log store expects no call
		330	+	assert.Contains(t, out.String(), "sweep the old runs")
		331	+	assert.Contains(t, out.String(), "kv closed")
		332	+	})
		333	+
		334	+	t.Run("a log directory that cannot be removed is logged and the orphans are still looked for", func(t *testing.T) {
		335	+	t.Parallel()
		336	+	u := newCIUnit(t, nil)
		337	+	ctx, out := logCapture(t)
		338	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return([]int64{3}, nil)
		339	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3}).Return(errors.New("permission denied")).Once()
		340	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
		341	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, domain.ErrNotFound)
		342	+	u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{9}).Return(errors.New("disk on fire")).Once()
		343	+	u.ci.Sweep(ctx)
		344	+	assert.Contains(t, out.String(), "remove the logs of the old runs")
		345	+	assert.Contains(t, out.String(), "permission denied")
		346	+	assert.Contains(t, out.String(), "remove the logs of runs that no longer exist")
		347	+	assert.Contains(t, out.String(), "disk on fire")
		348	+	})
		349	+
		350	+	t.Run("a listing of the log directories that fails is logged", func(t *testing.T) {
		351	+	t.Parallel()
		352	+	u := newCIUnit(t, nil)
		353	+	ctx, out := logCapture(t)
		354	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
		355	+	u.logs.EXPECT().RunIDs(mock.Anything).Return(nil, errors.New("io error"))
		356	+	u.ci.Sweep(ctx)
		357	+	assert.Contains(t, out.String(), "list the log directories")
		358	+	})
		359	+
		360	+	t.Run("a run with a lookup error keeps its logs", func(t *testing.T) {
		361	+	t.Parallel()
		362	+	u := newCIUnit(t, nil)
		363	+	ctx, out := logCapture(t)
		364	+	u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
		365	+	u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
		366	+	u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, errors.New("kv closed"))
		367	+	u.ci.Sweep(ctx) // RemoveRuns is not expected: an unknown state removes nothing
		368	+	assert.Contains(t, out.String(), "look up a run")
		369	+	})
		370	+	}
		371	+
		372	+	// M5: a log that would pass its limit appends nothing; the server adds a last
		373	+	// line saying so, fails the job and answers failed_precondition with the end.
		374	+	func TestLogLimitEndsTheJob(t *testing.T) {
		375	+	t.Parallel()
		376	+	u := newCIUnit(t, nil)
		377	+	j := ciJob(t, domain.CIPhaseRunning, jobSecret)
		378	+	j.Steps = []domain.CIStep{{Number: 1, State: domain.CIStepInProgress, StartedAt: now, HasLines: true}}
		379	+	u.expectToken(jobSecret, j)
		380	+	u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
		381	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(10, nil)
		382	+	u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(10), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
		383	+
		384	+	// stopForLogLimit: the run is loaded, the last line is forced in, the job ends.
		385	+	u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo, WorkflowYAML: "on: push\njobs:\n build:\n runs-on: x\n steps:\n - run: echo\n"}, nil)
		386	+	u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1, WorkflowPath: "w.yml"}, nil)
		387	+	u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{j}, nil)
		388	+	var forced []domain.CILogLine
		389	+	u.logs.EXPECT().AppendForced(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
		390	+	func(_ context.Context, _ ports.CILogRef, lines []domain.CILogLine) (int64, error) {
		391	+	forced = lines
		392	+	return 11, nil
		393	+	})
		394	+	var written ports.CIWrite
		395	+	u.store.EXPECT().Apply(mock.Anything, mock.Anything).RunAndReturn(
		396	+	func(_ context.Context, w ports.CIWrite) (uint64, error) { written = w; return 5, nil })
		397	+	u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPrivate}, nil)
		398	+	u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(11, nil).Maybe()
		399	+	u.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
		400	+
		401	+	_, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 10, []domain.CILogLine{{Time: now, Text: "one more"}})
		402	+	ended, ok := domain.IsJobEnded(err)
		403	+	require.True(t, ok, "%v", err)
		404	+	assert.Equal(t, domain.CIFailure, ended.Conclusion)
		405	+	assert.Contains(t, ended.Message, "512 MiB")
		406	+
		407	+	require.Len(t, forced, 1)
		408	+	assert.Contains(t, forced[0].Text, "##[error]")
		409	+	assert.Contains(t, forced[0].Text, "512 MiB")
		410	+	require.Len(t, written.Jobs, 1)
		411	+	assert.Equal(t, domain.CIPhaseEnded, written.Jobs[0].Phase)
		412	+	assert.Equal(t, domain.CIFailure, written.Jobs[0].Conclusion)
		413	+	assert.Equal(t, domain.CIStepCompleted, written.Jobs[0].Steps[0].State, "an open step is completed with the job")
		414	+	}

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


@@ -9,8 +9,6 @@ import (
9	9		"strconv"
10	10		"time"
11	11
12		-	"github.com/bnema/zerowrap"
13		-
14	12		"bnema.dev/ermite/internal/domain"
15	13		)
16	14

@@ -239,9 +237,11 @@ func (c *CI) checkQueue(ctx context.Context) {
239	237		}
240	238
241	239		// sweep removes the runs older than the retention, and the logs of runs that
242		-	// no longer exist (a repository deleted while the server was down).
	240	+	// no longer exist (a repository deleted while the server was down). Every
	241	+	// failure is logged and the next step still runs: a log directory that cannot
	242	+	// be removed is tried again at the next sweep.
243	243		func (c *CI) sweep(ctx context.Context) {
244		-	log := zerowrap.FromCtx(ctx)
	244	+	log := ctxLog(ctx)
245	245		ids, err := c.cfg.Store.Sweep(ctx, c.now().Add(-c.cfg.Retention))
246	246		if err != nil {
247	247		log.Error().Err(err).Msg("ci: sweep the old runs")

@@ -249,7 +249,7 @@ func (c *CI) sweep(ctx context.Context) {
249	249		}
250	250		if len(ids) > 0 {
251	251		if err := c.cfg.Logs.RemoveRuns(ctx, ids); err != nil {
252		-	log.Error().Err(err).Msg("ci: remove the logs of the old runs")
	252	+	log.Error().Err(err).Int("runs", len(ids)).Msg("ci: remove the logs of the old runs")
253	253		}
254	254		for _, id := range ids {
255	255		c.locks.Delete(id)

@@ -258,16 +258,22 @@ func (c *CI) sweep(ctx context.Context) {
258	258		}
259	259		onDisk, err := c.cfg.Logs.RunIDs(ctx)
260	260		if err != nil {
	261	+	log.Error().Err(err).Msg("ci: list the log directories")
261	262		return
262	263		}
263	264		var orphans []int64
264	265		for _, id := range onDisk {
265		-	if _, err := c.cfg.Store.GetRunSource(ctx, id); errors.Is(err, domain.ErrNotFound) {
	266	+	switch _, err := c.cfg.Store.GetRunSource(ctx, id); {
	267	+	case errors.Is(err, domain.ErrNotFound):
266	268		orphans = append(orphans, id)
	269	+	case err != nil:
	270	+	log.Error().Err(err).Int64("run_id", id).Msg("ci: look up a run with a log directory")
267	271		}
268	272		}
269	273		if len(orphans) > 0 {
270		-	_ = c.cfg.Logs.RemoveRuns(ctx, orphans)
	274	+	if err := c.cfg.Logs.RemoveRuns(ctx, orphans); err != nil {
	275	+	log.Error().Err(err).Int("runs", len(orphans)).Msg("ci: remove the logs of runs that no longer exist")
	276	+	}
271	277		}
272	278		}
273	279

 	"strconv"
 	"time"
-	"github.com/bnema/zerowrap"
 	"bnema.dev/ermite/internal/domain"
 )
 }
 // sweep removes the runs older than the retention, and the logs of runs that
-// no longer exist (a repository deleted while the server was down).
+// no longer exist (a repository deleted while the server was down). Every
+// failure is logged and the next step still runs: a log directory that cannot
+// be removed is tried again at the next sweep.
 func (c *CI) sweep(ctx context.Context) {
-	log := zerowrap.FromCtx(ctx)
+	log := ctxLog(ctx)
 	ids, err := c.cfg.Store.Sweep(ctx, c.now().Add(-c.cfg.Retention))
 	if err != nil {
 		log.Error().Err(err).Msg("ci: sweep the old runs")
 	}
 	if len(ids) > 0 {
 		if err := c.cfg.Logs.RemoveRuns(ctx, ids); err != nil {
-			log.Error().Err(err).Msg("ci: remove the logs of the old runs")
+			log.Error().Err(err).Int("runs", len(ids)).Msg("ci: remove the logs of the old runs")
 		}
 		for _, id := range ids {
 			c.locks.Delete(id)
 	}
 	onDisk, err := c.cfg.Logs.RunIDs(ctx)
 	if err != nil {
+		log.Error().Err(err).Msg("ci: list the log directories")
 		return
 	}
 	var orphans []int64
 	for _, id := range onDisk {
-		if _, err := c.cfg.Store.GetRunSource(ctx, id); errors.Is(err, domain.ErrNotFound) {
+		switch _, err := c.cfg.Store.GetRunSource(ctx, id); {
+		case errors.Is(err, domain.ErrNotFound):
 			orphans = append(orphans, id)
+		case err != nil:
+			log.Error().Err(err).Int64("run_id", id).Msg("ci: look up a run with a log directory")
 		}
 	}
 	if len(orphans) > 0 {
-		_ = c.cfg.Logs.RemoveRuns(ctx, orphans)
+		if err := c.cfg.Logs.RemoveRuns(ctx, orphans); err != nil {
+			log.Error().Err(err).Int("runs", len(orphans)).Msg("ci: remove the logs of runs that no longer exist")
+		}
 	}
 }

M internal/usecase/export_test.go +8 -1


@@ -2,7 +2,14 @@
2	2
3	3		package usecase
4	4
5		-	import "bnema.dev/ermite/internal/domain"
	5	+	import (
	6	+	"context"
	7	+
	8	+	"bnema.dev/ermite/internal/domain"
	9	+	)
6	10
7	11		// AuthorOf exposes authorOf to the tests.
8	12		func AuthorOf(p domain.Principal) string { return authorOf(p) }
	13	+
	14	+	// Sweep exposes the daily sweep to the tests.
	15	+	func (c *CI) Sweep(ctx context.Context) { c.sweep(ctx) }

 package usecase
-import "bnema.dev/ermite/internal/domain"
+import (
+	"context"
+	"bnema.dev/ermite/internal/domain"
+)
 // AuthorOf exposes authorOf to the tests.
 func AuthorOf(p domain.Principal) string { return authorOf(p) }
+// Sweep exposes the daily sweep to the tests.
+func (c *CI) Sweep(ctx context.Context) { c.sweep(ctx) }